package cn.doitedu.analy;

import cn.doitedu.analy.groupon.GrouponEntity;
import cn.doitedu.analy.groupon.GrouponMapper;
import cn.doitedu.analy.groupon.GrouponReducer;
import cn.doitedu.analy.groupon.GrouponSink;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import utils.TransferUtil;

import java.util.Properties;
import java.util.TimerTask;

/**
 * 读取kafka中的bin-log日志数据
 * 本代码的写法，只能是对每一个表来单独写一个任务，因为输出的sink只能定死目标topic
 * 可以改造成侧流输出
 */
public class GrouponAnaly {

    private final static String sourceTopic = "order_info";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties props = new Properties();
        props.setProperty("bootstrap.servers","h1:9092,h2:9092,h3:9092");
        props.setProperty("group.id","doitedu");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test1", new SimpleStringSchema(), props);
        DataStreamSource<String> ds = env.addSource(consumer);
        env.enableCheckpointing(5000);
        ds.print();

        SingleOutputStreamOperator<GrouponEntity> map = ds.flatMap(new GrouponMapper());
        SingleOutputStreamOperator<GrouponEntity> times = map.keyBy("userId").timeWindowAll(Time.seconds(2)).reduce(new GrouponReducer());

        // 结果写入hbase
        times.addSink(new GrouponSink());

        env.execute("TransferData");
    }
}
